package com.test.study.ToolsUtils.kafka;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.junit.Test;

/**
 * 本类用于Kafka的消费者测试
 * 
 * @author huyong
 *
 */
public class KafkaConsumerServer {

	private KafkaConsumer<String, String> consumer;

	/**
	 * kafka的连接是以tcp协议连接的 用完后需要将资源释放
	 * Kafka为分区中的每条记录维护一个数字偏移量。此偏移量充当该分区内记录的唯一标识符，并且还表示消费者在分区中的位置。例如，位于第5位的消费者已经消耗了偏移0到4的记录，并且接下来将接收具有偏移5的记录。实际上有两个与消费者的用户相关的位置概念：
	 * 消费者的位置给出了下一个记录的偏移量。它将比消费者在该分区中看到的最高偏移量大一个。每次消费者在轮询调用中接收消息时，它会自动前进（持续时间）。
	 * 提交的位置是安全存储的最后一个偏移量。如果进程失败并重新启动，则这是消费者将恢复到的偏移量。消费者可以定期自动提交抵消;或者它可以选择通过调用其中一个提交API（例如commitSync和commitAsync）手动控制此提交位置。
	 * 
	 * 
	 * 本方法是自动偏移提交
	 * 
	 * 
	 * 
	 * huyong 2019年3月9日
	 */
	@SuppressWarnings("resource")
	@Test
	public void ConsumerServer01() {
		String topic1 = "hello";
		String topic2 = "bar";
		Properties props = new Properties();
		props.put("bootstrap.servers", "localhost:9092");
		props.put("group.id", "test-server-huyong");
		props.put("enable.auto.commit", "true");// 设置enable.auto.commit意味着自动提交偏移量，其频率由config auto.commit.interval.ms控制。
		props.put("auto.commit.interval.ms", "1000");//
		// 解串器设置指定如何将字节转换为对象。 例如，通过指定字符串反序列化器，我们说我们的记录的键和值只是简单的字符串。
		props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		// session.timeout.ms的持续时间内崩溃或无法发送心跳，则消费者将被视为已死，并且将重新分配其分区。
		// 消费者也可能遇到“活锁”情况，即继续发送心跳，但没有取得进展。为防止消费者在这种情况下无限期地占用其分区，我们使用max.poll.interval.ms设置提供活动检测机制。基本上，如果您不至少像配置的最大间隔那样频繁地调用轮询，则客户端将主动离开该组，以便其他使用者可以接管其分区。发生这种情况时，您可能会看到偏移提交失败（由对commitSync（）的调用抛出的CommitFailedException指示）。这是一种安全机制，可确保只有组中的活动成员才能提交偏移量。因此，要留在群组中，您必须继续进行选举
		// max.poll.interval.ms：通过增加预期轮询之间的间隔，您可以为消费者提供更多时间来处理从poll（Duration）返回的一批记录。缺点是增加此值可能会延迟组重新平衡，因为消费者只会加入轮询调用中的重新平衡。您可以使用此设置来限制完成重新平衡的时间，但如果消费者实际上不能经常调用轮询，则可能会导致进度降低。
		// max.poll.records：使用此设置可限制从单个调用返回到poll的总记录数。这可以更容易地预测每个轮询间隔内必须处理的最大值。通过调整此值，您可以减少轮询间隔，这将减少组重新平衡的影响。
		KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
		// 使用者将订阅主题foo和bar作为一组使用group.id配置的称为test-server-huyong的消费者的一部分。
		// Map<String,Integer> topicMap=new HashMap<String,Integer>();//这个Map的意义：
		// key:表示那个topic value：表示我在这个进程中起几个线程去消费这个topic 一般和partition的个数相同
		consumer.subscribe(Arrays.asList(topic1, topic2));
		while (true) {
			ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(100));
			for (ConsumerRecord<String, String> record : records)
				System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
		}
	}

	/**
	 * 用户还可以控制何时应将记录视为已消耗并因此提交其偏移量，而不是依赖于消费者定期提交消耗的偏移量。
	 * 当消息的消耗与某些处理逻辑耦合时，这是有用的，因此消息在完成处理之前不应被视为已消耗。
	 * 我们将使用一批记录并在内存中批量处理它们。当我们有足够的记录批量时，我们会将它们插入到数据库中。如果我们允许偏移量自动提交，如上例所示，则记录将在轮询中返回给用户后被视为已消耗。然后，我们的进程可能在批处理记录之后但在将它们插入数据库之前失败。
	 * 为避免这种情况，我们将仅在将相应的记录插入数据库后手动提交偏移量。这使我们能够准确控制何时考虑使用记录。这提出了相反的可能性：进程可能在插入数据库之后但在提交之前的时间间隔内失败（即使这可能只是几毫秒，这是可能的）。在这种情况下，接管消耗的过程将消耗最后提交的偏移量，并将重复插入最后一批数据。以这种方式使用Kafka提供通常称为“至少一次”的传递保证，因为每个记录可能会被传递一次但是在失败的情况下可能会被复制。
	 * huyong 2019年3月9日
	 */
	@Test
	public void consumerServer02() {
		Properties props = new Properties();
		props.put("bootstrap.servers", "localhost:9092");
		props.put("group.id", "test");
		props.put("enable.auto.commit", "false");// 取消自动提交功能
		props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		consumer = new KafkaConsumer<String, String>(props);
		consumer.subscribe(Arrays.asList("hello", "bar"));
		final int minBatchSize = 200;
		List<ConsumerRecord<String, String>> buffer = new ArrayList<ConsumerRecord<String, String>>();
		while (true) {
			Duration timeout = Duration.ofSeconds(100);
			ConsumerRecords<String, String> records = consumer.poll(timeout);
			for (ConsumerRecord<String, String> record : records) {
				System.out.println(String.format("key:%s,partition:%s,offset:%s,value:%s", record.key(),
						record.partition(), record.offset(), record.value()));
				buffer.add(record);
			}
			if (buffer.size() >= minBatchSize) {
				insertIntoDb(buffer);
				consumer.commitSync();
				buffer.clear();
			}
		}

	}

	private void insertIntoDb(List<ConsumerRecord<String, String>> buffer) {
		if (null != buffer && buffer.size() > 0) {
			ConsumerRecord<String, String> consumerRecord = buffer.get(0);
			System.out.println(String.format("key:%s,partition:%s,offset:%s,value:%s", consumerRecord.key(),
					consumerRecord.partition(), consumerRecord.offset(), consumerRecord.value()));
		}

	}
	/**
	 * 
	 * huyong
	 * 2019年3月11日
	 */
	public void consumerServer03() {
//	    try {
//	         while(running) {
//	             ConsumerRecords<String, String> records = consumer.poll(Long.MAX_VALUE);
//	             for (TopicPartition partition : records.partitions()) {
//	                 List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
//	                 for (ConsumerRecord<String, String> record : partitionRecords) {
//	                     System.out.println(record.offset() + ": " + record.value());
//	                 }
//	                 long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
//	                 //提交的偏移量应始终是应用程序将读取的下一条消息的偏移量。 因此，在调用commitSync（偏移量）时，您应该在最后处理的消息的偏移量中添加一个。
//	                 consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset + 1)));
//	             }
//	         }
//	     } finally {
//	       consumer.close();
//	     }
	 
	}
	
	

}
